// 定义包路径，标识该类所在的包结构
package bigdata.hermesfuxi.eagle.etl.functions;

// 导入阿里巴巴的FastJSON库，用于JSON字符串与Java对象之间的转换
import com.alibaba.fastjson.JSON;
// 导入Flink的类型信息类，用于类型推断（虽然在此类中未直接使用，但可能用于类型系统）
// 导入Flink的ProcessFunction抽象类，用于处理流数据中的每个元素
import org.apache.flink.streaming.api.functions.ProcessFunction;
// 导入Flink的Collector接口，用于将处理后的数据发送到下游
import org.apache.flink.util.Collector;

// 导入Java反射包中的ParameterizedType接口，用于获取泛型类型信息
import java.lang.reflect.ParameterizedType;

// 定义泛型类JsonToBeanFunc，继承自ProcessFunction
// 泛型T表示要转换的目标Java对象类型
// 输入类型为String（JSON字符串），输出类型为T（目标Java对象）
public class JsonToBeanFunc<T> extends ProcessFunction<String, T> {
    // 该类继承自Flink的ProcessFunction，用于将JSON字符串转换为Java对象
    // 泛型T表示目标Java对象的类型
    // 无需额外成员变量，所有逻辑在processElement方法中实现

    // 重写父类的processElement方法，这是处理流数据元素的核心方法
    @Override
    public void processElement(String value, Context ctx, Collector<T> out) throws Exception {
        // 使用try-catch块捕获可能的异常，确保程序稳定性
        try{
            // 抑制未检查的类型转换警告，因为泛型类型擦除导致的类型转换
            @SuppressWarnings("unchecked")
            // 通过反射获取泛型T的实际Class类型
            // 1. getClass()获取当前类的Class对象
            // 2. getGenericSuperclass()获取父类的泛型类型信息
            // 3. 强制转换为ParameterizedType以获取泛型参数
            // 4. getActualTypeArguments()[0]获取第一个泛型参数（即T的类型）
            // 5. 强制转换为Class<T>类型
            Class<T> tClass = (Class<T>)((ParameterizedType)getClass().getGenericSuperclass()).getActualTypeArguments()[0];
            // 使用FastJSON的parseObject方法将JSON字符串转换为指定类型的Java对象
            // value是输入的JSON字符串，tClass是目标对象的Class类型
            T t = JSON.parseObject(value, tClass);
            // 检查转换后的对象是否为空，避免发送null值到下游
            if(t != null){
                // 将转换成功的Java对象发送到下游处理
                out.collect(t);
            }
        }catch (Exception e){
            // 捕获所有异常并打印堆栈跟踪，便于调试和问题定位
            e.printStackTrace();
        }
    }
}
